package TableAndSQL

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors.{Csv, Kafka, Schema}

object SourceFromKafka {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //创建一个基于blink的流式
    val BlinkStreamSetting = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val blinkStreamTableEnv = StreamTableEnvironment.create(env, BlinkStreamSetting)

    blinkStreamTableEnv.connect(new Kafka()
      .version("0.11")
      .topic("sensor")
      .property("zookeeper.connect", "master:2181")
      .property("bootstrap.servers", "master:9092")
    )
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      )
      .createTemporaryTable("kafka")
    val inputTable = blinkStreamTableEnv.from("kafka")

    val resultTable = inputTable
      .select('id, 'temperature)
      .filter('id === "sensor_1")

    resultTable.toAppendStream[(String, Double)].print()
    env.execute()
  }
}
